package com.navinfo.platform.trip.analysis.flink.function;

import com.navinfo.platform.trip.analysis.flink.config.SystemConfig;
import com.navinfo.platform.trip.analysis.pojo.AlgorithmChainData;
import com.navinfo.platform.trip.analysis.pojo.AlgorithmOuterData;
import com.navinfo.platform.trip.analysis.pojo.DF_OuterStatisticData;
import com.navinfo.platform.trip.analysis.pojo.PreTripInfo;
import com.navinfo.platform.trip.analysis.service.AlgorithmChainService;
import com.navinfo.platform.trip.analysis.service.ServiceFactory;
import com.navinfo.platform.trip.analysis.service.TripJudgeService;
import com.navinfo.platform.trip.analysis.service.impl.common.Algorithm0200ChainServiceImpl;
import com.navinfo.platform.trip.analysis.util.TripStatisticUtils;
import com.navinfo.platform.trip.common.arithmetic.data.DrivingData;
import com.navinfo.platform.trip.common.arithmetic.data.OuterEventData;
import com.navinfo.platform.trip.common.config.ConfigMap;
import com.navinfo.platform.trip.common.enums.RunningTypeEnum;
import com.navinfo.platform.trip.common.enums.SystemTypeEnum;
import com.navinfo.platform.trip.common.pojo.Point;
import com.navinfo.platform.trip.common.pojo.PointProtocol;
import com.navinfo.platform.trip.common.util.CommonUtils;
import com.navinfo.platform.trip.common.util.DateUtils;
import com.navinfo.platform.trip.common.vehicle.VehicleInfoService;
import com.navinfo.platform.trip.common.vehicle.VehicleInfoServiceImpl;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * 密集0F37和位置点0200双流Union后的处理函数<br/>
 * 使用窗口中的0F37和0200进行行程统计<br/>
 * 输出行程统计信息OuterStatisticData<br/>
 * 旁路输出行程事件信息OuterEventData<br/>
 * @author: web
 * @data:  2020-04-10
 **/
public class TripFunction extends ProcessWindowFunction<PointProtocol, DF_OuterStatisticData, Long, GlobalWindow> {
    private static final Logger logger = LoggerFactory.getLogger(TripFunction.class);

    /*
     *行程时长小于多少秒不保存，单位s
     */
    private int tripMinDuration=120;

    /**
     * 车辆配置信息服务
     */
    private VehicleInfoService vehicleService;
    /**
     * 行程切分服务
     */
    private TripJudgeService tripJudgeService;

    /**
     * 使用位置点0200和密集0F37数据进行的驾驶行为分析
     */
    private AlgorithmChainService algorithmChain;
    /**
     * 使用位置点0200数据进行的单独事件收集算法
     */
    private AlgorithmChainService algorithm0200Chain;

    /**
     * 旁路输出的标签
     */
    private OutputTag<OuterEventData> eventOutputTag;

    /**
     * 行程统计的缓存数据
     */
    private final ValueStateDescriptor<DrivingData> tripStateDesc = new ValueStateDescriptor<>("trip_state", TypeInformation.of(DrivingData.class));

    public TripFunction(OutputTag<OuterEventData> eventOutputTag) {
        this.eventOutputTag = eventOutputTag;
    }

    @Override
    @SuppressWarnings("unchecked")
    public void open(Configuration parameters) throws Exception {
        Map jobMap = getRuntimeContext().getExecutionConfig().getGlobalJobParameters().toMap();

        //系统类型
        SystemTypeEnum systemTypeEnum = SystemTypeEnum.valueOf(Integer.parseInt((String) jobMap.get(SystemConfig.systemTypeKey)));

        Optional.ofNullable(jobMap.get(SystemConfig.tripMinDurationKey)).ifPresent(e -> tripMinDuration=Integer.parseInt(e.toString()));

        //初始化算法的配置
        ConfigMap.init(jobMap);

        //初始化算法链服务
        algorithmChain = ServiceFactory.singleServiceImpl(systemTypeEnum, AlgorithmChainService.class);
        algorithm0200Chain = new Algorithm0200ChainServiceImpl();

        //初始化车辆信息服务
        vehicleService = new VehicleInfoServiceImpl();
        vehicleService.init(jobMap);

        //@todo 初始化打分服务

        //初始化行程切分服务
        tripJudgeService = ServiceFactory.singleServiceImpl(systemTypeEnum, TripJudgeService.class);

        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    /**
     * 使用窗口中的0F37和0200进行行程统计<br/>
     * 输出行程统计信息OuterStatisticData<br/>
     * 旁路输出行程事件信息OuterEventData<br/>
     * @param tid 当前Key
     * @param ctx 上下文
     * @param iterable  合并的0F37和0200数据
     * @param out 输出
     * @throws Exception 异常
     */
    @Override
    public void process(Long tid, Context ctx, Iterable<PointProtocol> iterable, Collector<DF_OuterStatisticData> out) throws Exception {
        //0200和0F37点整体排序
        List<PointProtocol> points = StreamSupport.stream(iterable.spliterator(), false).sorted(Point.COMPARATOR).collect(Collectors.toList());

        //构建算法链的缓存对象
        ValueState<DrivingData> tripDrivingStore = ctx.globalState().getState(tripStateDesc);
        DrivingData drivingData = Optional.ofNullable(tripDrivingStore.value()).orElse(new DrivingData());
        drivingData.setVehicleInfo(vehicleService.get(tid));
        logger.debug("加载算法链的中间数据drivingData:{}", drivingData);

        //@todo 设置瓦片信息，流计算不使用瓦片信息计算经过城市

        //使用位置点切分行程
        List<PreTripInfo> preTripInfos = tripJudgeService.judgeTrip(points, drivingData);
        if(CommonUtils.isCollectionNotEmpty(preTripInfos)){
            //构建算法链的执行对象
            AlgorithmChainData chainData = new AlgorithmChainData();
            chainData.setTid(tid);
            chainData.setDrivingData(drivingData);
            chainData.setRunningType(RunningTypeEnum.STREAM);

            //执行每个行程的算法链
            for(int i=0, size=preTripInfos.size(); i<size; i++){
                PreTripInfo preTripInfo = preTripInfos.get(i);

                //跨天标志处理，批计算：一天当中的最后一个行程设值为true；流计算：开始时间和结束时间不是同一天设值为true
                if(RunningTypeEnum.STREAM.equals(chainData.getRunningType())){
                    preTripInfo.setHandleCrossDay(!DateUtils.sameDay(preTripInfo.getStartTime(), preTripInfo.getEndTime()));
                }else{
                    if(i == size-1){
                        preTripInfo.setHandleCrossDay(true);
                    }
                }

                //设置当前行程信息
                chainData.setPreTripInfo(preTripInfo);
                chainData.setTripDay(new Date(preTripInfo.getStartTime()));

                //执行驾驶行为分析算法链
                algorithmChain.execute(chainData);

                //执行0200单独事件收集算法链
                algorithm0200Chain.execute(chainData);
            }

            //批计算：在当天所有行程处理完后，进行精准里程油耗、跨天的处理
            if(RunningTypeEnum.BATCH.equals(chainData.getRunningType())) {
                //跨天处理
                TripStatisticUtils.dealBatchCrossDay(chainData.getOuterData().getOuterStatistics(), drivingData);
            }

            //结果输出处理
            AlgorithmOuterData outerData = chainData.getOuterData();

            //输出统计信息
            outerData.getOuterStatistics().forEach(e -> {
                //过滤小于tripEndTime的行程
                if(e.getTripDuration() >= tripMinDuration){
                    DF_OuterStatisticData statisticData = new DF_OuterStatisticData();
                    statisticData.setTid(e.getTid());
                    statisticData.setTripDay(e.getDay());
                    statisticData.setTripId(e.getTripId());
                    statisticData.setRouteStartTime(e.getRouteStartTime());
                    statisticData.setRouteStartLatitude(e.getRouteStartLatitude());
                    statisticData.setRouteStartLongitude(e.getRouteStartLongitude());
                    statisticData.setRouteStartHeight(e.getRouteStartHeight());
                    statisticData.setRouteStartDirection(e.getRouteStartDirection());
                    statisticData.setRouteStartMileage(e.getRouteStartMileage());
                    statisticData.setRouteStartFuel(e.getRouteStartFuel());
                    statisticData.setRouteEndTime(e.getRouteEndTime());
                    statisticData.setRouteEndLatitude(e.getRouteEndLatitude());
                    statisticData.setRouteEndLongitude(e.getRouteEndLongitude());
                    statisticData.setRouteEndHeight(e.getRouteEndHeight());
                    statisticData.setRouteEndDirection(e.getRouteEndDirection());
                    statisticData.setRouteEndMileage(e.getRouteEndMileage());
                    statisticData.setRouteEndFuel(e.getRouteEndFuel());

                    statisticData.setRouteIntegralMileage(e.getRouteIntegralMileage());
                    statisticData.setRouteIntegralFuel(e.getRouteIntegralFuel());
                    statisticData.setRouteIntegralDuration(e.getRouteIntegralDuration());

                    statisticData.setTripMileage(e.getTripMileage());
                    statisticData.setTripFuel(e.getTripFuel());
                    statisticData.setTripDuration(e.getTripDuration());
                    statisticData.setTripPower(e.getTripPower());
                    statisticData.setPreciseFlag(e.getPreciseFlag());
                    statisticData.setSType(e.getSourceType());
                    statisticData.setCrossDayFlag(e.getCrossDayFlag());
                    statisticData.setInterCrossDay(e.getInterCrossDay());

                    statisticData.setSharpTurningNumber(e.getSharpTurningNumber_DFN());
                    statisticData.setNightRunningNumber(e.getNightRunningNumber_DFN());
                    statisticData.setEmergencyStopNumber(e.getEmergencyStopNumber_DFN());
                    statisticData.setFlameoutCoastNumber(e.getFlameoutCoastNumber_DF());

                    statisticData.setColdRunningMileage(e.getColdRunningMileage_DFN());
                    statisticData.setColdRunningFuel(e.getColdRunningFuel_DFN());
                    statisticData.setColdRunningDuration(e.getColdRunningDuration_DFN());
                    statisticData.setColdRunningNumber(e.getColdRunningNumber_DFN());

                    statisticData.setNoEcoDrivingMileage(e.getNoEcoMileage_DFN());
                    statisticData.setNoEcoDrivingFuel(e.getNoEcoFuel_DFN());
                    statisticData.setNoEcoDrivingDuration(e.getNoEcoDuration_DFN());
                    statisticData.setNoEcoDrivingNumber(e.getNoEcoNumber_DFN());
                    statisticData.setNoEcoDrivingProportion(e.getNoEcoProportion_DFN());
                    statisticData.setFatigueDrivingDuration(e.getFatigueDrivingDuration());
                    statisticData.setFatigueDrivingNumber(e.getFatigueDrivingNumber());

                    statisticData.setFuel100KM(e.fuel100KM());
                    statisticData.setSharpUpSpeedNumber100KM(e.sharpUpSpeedNumber100KM());
                    statisticData.setSharpDownSpeedNumber100KM(e.sharpDownSpeedNumber100KM());
                    statisticData.setSharpTurningNumber100KM(e.sharpTurningNumber100KM());
                    statisticData.setEmergencyStopNumber100KM(e.emergencyStopNumber100KM());
                    statisticData.setFlameOutCoastNumber100KM(e.flameoutCoastNumber100KM());
                    statisticData.setSpeedOverDurationHourly(e.speedOverDurationHourly());
                    statisticData.setLongIdleDurationHourly(e.longIdleDurationHourly());
                    statisticData.setLargeGasDurationHourly(e.largeGasDurationHourly());
                    statisticData.setNoEcoDurationHourly(e.noEcoDurationHourly());
                    statisticData.setNeutralGearCoastDurationHourly(e.neutralGearCoastDurationHourly());
                    statisticData.setFatigueDrivingDurationHourly(e.fatigueDrivingDurationHourly());
                    statisticData.setNightRunningDurationHourly(e.nightRunningDurationHourly());

                    out.collect(statisticData);
                }
            });

            //旁路输出事件信息
            outerData.getOuterEvents().forEach(e -> ctx.output(eventOutputTag, e));

            //更新算法链的缓存对象
            DrivingData outDrivingData = outerData.getOuterDrivingData();
            if (null != outDrivingData) {
                outDrivingData.setVehicleInfo(null);
                outDrivingData.setTileLcMapping(null);
            }
            tripDrivingStore.update(outDrivingData);
        }
    }
}
